package com.cnebula.common.hadoop.oozie.client.impl;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;

import org.apache.http.NameValuePair;
import org.apache.http.ParseException;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.util.EntityUtils;
import org.osgi.service.component.ComponentContext;

import com.cnebula.common.annotations.es.ESRef;
import com.cnebula.common.annotations.es.EasyService;
import com.cnebula.common.conf.IEasyServiceConfAdmin;
import com.cnebula.common.hadoop.oozie.client.Configuration;
import com.cnebula.common.hadoop.oozie.client.IOozieRestService;
import com.cnebula.common.hadoop.oozie.client.OozieConf;
import com.cnebula.common.hadoop.oozie.client.Property;
import com.cnebula.common.log.ILog;
import com.cnebula.common.xml.IEasyObjectXMLTransformer;

@EasyService
public class OozieRestServiceImpl implements IOozieRestService {

	@ESRef
	ILog log;

	@ESRef
	IEasyServiceConfAdmin confAdmin;

	@ESRef
	IEasyObjectXMLTransformer xtf;

	OozieConf conf;

	OozieHttpClient oozieHttpClient;

	protected void activate(ComponentContext ctx) throws Exception {
		conf = confAdmin.get(OozieConf.class.getName(), OozieConf.class);
		if (conf == null) {
			log.error("[OozieConf] 加载失败，请检查相关的配置是否正确。");
			return;
		}
		if (conf.host == null || conf.host.trim().equals("")) {
			log.error("[OozieConf] host 加载失败，请检查相关的配置是否正确。");
			return;
		}
		oozieHttpClient = new OozieHttpClient();
	}

	public String submitJob(List<Property> properties, List<NameValuePair> params, String version) {
		StringBuilder url = new StringBuilder("http://" + conf.host + ":" + conf.port);
		if (version == null) {
			version = "v1";
		}
		url.append("/oozie/" + version + "/jobs");
		try {
			String str = EntityUtils.toString(new UrlEncodedFormEntity(params));
			if (str != null && !str.trim().equals("")) {
				url.append("?");
				url.append(str);
			}
		} catch (ParseException e) {
			e.printStackTrace();
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		Configuration configuration = new Configuration();
		configuration.setProperties(properties);
		String data = xtf.export(configuration);
		String rst = oozieHttpClient.post(url.toString(), data);
		return rst;
	}

	public String submitStandardJob(List<Property> properties, boolean immediate) {
		List<NameValuePair> params = new ArrayList<NameValuePair>();
		if (immediate) {
			params.add(new BasicNameValuePair("action", "start"));
		}
		String rst = submitJob(properties, params, null);
		return rst;
	}

	public String submitMapReduceJob(List<Property> properties, boolean immediate) {
		List<NameValuePair> params = new ArrayList<NameValuePair>();
		if (immediate) {
			params.add(new BasicNameValuePair("action", "start"));
		}
		params.add(new BasicNameValuePair("jobtype", "mapreduce"));

		String rst = submitJob(properties, params, null);
		return rst;
	}

	public String submitPigJob(List<Property> properties, boolean immediate) {
		List<NameValuePair> params = new ArrayList<NameValuePair>();
		if (immediate) {
			params.add(new BasicNameValuePair("action", "start"));
		}
		params.add(new BasicNameValuePair("jobtype", "pig"));

		String rst = submitJob(properties, params, null);
		return rst;
	}

	public String submitHiveJob(List<Property> properties, boolean immediate) {
		List<NameValuePair> params = new ArrayList<NameValuePair>();
		if (immediate) {
			params.add(new BasicNameValuePair("action", "start"));
		}
		params.add(new BasicNameValuePair("jobtype", "hive"));

		String rst = submitJob(properties, params, null);
		return rst;
	}

	public String getJobInformation(String jobId) {
		if (jobId == null || jobId.trim().equals("")) {
			return null;
		}

		StringBuilder url = new StringBuilder("http://" + conf.host + ":" + conf.port);
		url.append("/oozie/v1/job");
		url.append("/" + jobId);
		url.append("?show=info&timezone=GMT");
		String rst = oozieHttpClient.get(url.toString());
		return rst;
	}

	
	public String killStandardJob(List<Property> properties, boolean immediate) {
		List<NameValuePair> params = new ArrayList<NameValuePair>();
		if (immediate) {
			params.add(new BasicNameValuePair("action", "kill"));
		}
		String rst = submitJob(properties, params, null);
		return rst;
	}
	
}
